package day3

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

import java.time.Instant

object time_consume {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\新建文件夹\\HCIA初级\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)
    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))

    sc.textFile("D:\\新建文件夹\\HCIA初级\\240722\\files\\consume.log")
      //默认RDD类型: (String,(String,String))
      .map( i => {
        val list = i.split("\t")
        ( list(1),( list(2).toLong,BigDecimal(list(3)) ) )
      })
      .sortBy(_._2._1)
      .groupByKey()
      /* 得到一个数据：
       * Seq((1594003200,234.56), (1594003200,123.45), (1619942400,345.67)
       * 通过RDD类型判断，可以得知这是一个Iterable(迭代器)
       * 通过输出能够看到，这个迭代器中内容是一个Seq集合
       */
      .mapValues(i=>{
        val (time,money) = i.unzip
        val time_string = time.map(tmp => {
          Instant.ofEpochSecond(tmp).toString
        })
        //返回的value，时间戳-->年月日表达，金额求和
        (time_string,money.sum)
      })
      .sortByKey()
      //得到的PairRDD：(name,(time,money))
      .foreach(i=>{
        println("名字：" + i._1)
        i._2._1.foreach(i => print(i+"\t"))
        println("\n消费总和：" + i._2._2)
        println("************************************")
      })

  }
}